import com.vividsolutions.jts.geom.Envelope;
import com.vividsolutions.jts.geom.Point;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.datasyslab.geospark.spatialOperator.JoinQuery;
import org.datasyslab.geospark.spatialRDD.PointRDD;
import org.datasyslab.geospark.spatialRDD.RectangleRDD;
import org.neo4j.gis.spatial.Layer;
import org.neo4j.gis.spatial.SpatialDatabaseService;
import scala.Tuple2;
import scala.tools.jline.internal.Log;

import java.io.IOException;
import java.io.InputStream;
import java.util.HashSet;
import java.util.List;
import java.util.Properties;

/**
 * Created by Dell on 2016/12/11 0011.
 */
public class JavaWordCount {

    public static JavaSparkContext sc;
    static Properties prop;
    static InputStream input;
    static String InputLocation;
    static String InputLocationQueryWindow;
    static Integer offset;
    static String splitter;
    static String gridType;
    static String indexType;
    static Integer numPartitions;

    public static void onceExecutedBeforeAll() {
        SparkConf conf = new SparkConf().setAppName("JavaWordCount").setMaster("local[*]")
//                .setMaster("spark://192.168.0.113:7077")
                .setJars(new String[]{"C:\\Users\\Dell\\IdeaProjects\\sparkSpatial\\out\\artifacts\\sparkSpatial_jar\\sparkSpatial.jar"})
                .set("spark.executor.memory", "8g");
        sc = new JavaSparkContext(conf);
        Logger.getLogger("org").setLevel(Level.WARN);
        Logger.getLogger("akka").setLevel(Level.WARN);
        prop = new Properties();
        input = JavaWordCount.class.getClassLoader().getResourceAsStream("point.test.properties");
        InputLocation = "file://" + JavaWordCount.class.getClassLoader().getResource("primaryroads.csv").getPath();
        offset = 0;
        splitter = "";
        gridType = "";
        indexType = "";
        numPartitions = 0;

        try {
            // load a properties file
            prop.load(input);
            InputLocation = "file://" + JavaWordCount.class.getClassLoader().getResource(prop.getProperty("inputLocation")).getPath();
            InputLocationQueryWindow = "file://" + JavaWordCount.class.getClassLoader().getResource(prop.getProperty("queryWindowSet")).getPath();
            offset = Integer.parseInt(prop.getProperty("offset"));
            splitter = prop.getProperty("splitter");
            gridType = prop.getProperty("gridType");
            indexType = prop.getProperty("indexType");
            numPartitions = Integer.parseInt(prop.getProperty("numPartitions"));

        } catch (IOException ex) {
            ex.printStackTrace();
        } finally {
            if (input != null) {
                try {
                    input.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public static void testSpatialJoinQuery() throws Exception {

        RectangleRDD rectangleRDD = new RectangleRDD(sc, InputLocationQueryWindow, offset, splitter, numPartitions);

        PointRDD pointRDD = new PointRDD(sc, InputLocation, offset, splitter, gridType, numPartitions);

        JoinQuery joinQuery = new JoinQuery(sc, pointRDD, rectangleRDD);

        List<Tuple2<Envelope, HashSet<Point>>> result = joinQuery.SpatialJoinQuery(pointRDD, rectangleRDD).collect();
        assert result.get(1)._1().getUserData() != null;
        for (int i = 0; i < result.size(); i++) {
            if (result.get(i)._2().size() != 0) {
                assert result.get(i)._2().iterator().next().getUserData() != null;
            }
        }
    }

    public static void main(String[] args) throws Exception {
        Logger.getLogger("org").setLevel(Level.WARN);
        Logger.getLogger("akka").setLevel(Level.WARN);
        onceExecutedBeforeAll();
        try {
            testSpatialJoinQuery();
        } catch (Exception e) {
            Log.error(e);
        } finally {
            sc.stop();
        }
    }
}
